package link.jfire.socket.socketserver.handler;

import java.nio.ByteBuffer;
import java.util.Arrays;
import link.jfire.baseutil.collection.ByteBufferPool;
import link.jfire.baseutil.collection.ByteCache;
import link.jfire.baseutil.order.AescComparator;
import link.jfire.baseutil.order.DescComparator;
import link.jfire.baseutil.simplelog.ConsoleLogFactory;
import link.jfire.baseutil.simplelog.Logger;
import link.jfire.socket.socketserver.exception.ServerStopException;
import link.jfire.socket.socketserver.interceptor.EntryInterceptor;
import link.jfire.socket.socketserver.storage.Entry;
import link.jfire.socket.socketserver.storage.ringarray.RingArray;
import link.jfire.socket.socketserver.util.HeadFactory;

/**
 * 消息动作类，完成对消息的进入拦截，请求处理，出口拦截，写出到通道。实现Callable接口，放入线程池进行调用。
 * 
 * @author linbin
 * 
 */
public final class MessageAction implements Runnable
{
    private volatile long      cursor = 0;
    private Logger             logger = ConsoleLogFactory.getLogger();
    private EntryInterceptor[] inInterceptors;
    private EntryInterceptor[] outInterceptors;
    private int                interceptorNum;
    // 命令数组
    private byte[]             command;
    // 命令数组的长度
    private int                limit;
    private EntryHandler[]     handlers;
    private RingArray          ringArray;
    private HeadFactory        headFactory;
    
    /**
     * 使用拦截器数组初始化对象，会自动将拦截器进行排序，按照入口顺序和出口顺序存放
     * 
     * @param interceptors
     */
    public MessageAction(EntryInterceptor[] interceptors, EntryHandler[] handlers, RingArray ringArray, HeadFactory headFactory)
    {
        this.inInterceptors = interceptors;
        Arrays.sort(this.inInterceptors, new AescComparator());
        interceptorNum = inInterceptors.length;
        outInterceptors = Arrays.copyOf(inInterceptors, interceptorNum);
        Arrays.sort(outInterceptors, new DescComparator());
        this.handlers = handlers;
        limit = handlers.length;
        command = new byte[limit];
        for (int i = 0; i < limit; i++)
        {
            command[i] = handlers[i].interestedDataPacketType();
        }
        this.headFactory = headFactory;
        this.ringArray = ringArray;
    }
    
    @Override
    public void run()
    {
        Entry entry;
        while (true)
        {
            if (ringArray.isAvailable(cursor) == false)
            {
                try
                {
                    logger.debug("等待序号:{}", cursor);
                    ringArray.waitFor(cursor);
                }
                catch (ServerStopException e)
                {
                    logger.error("停止");
                    break;
                }
            }
            entry = ringArray.entryAt(cursor);
            try
            {
                if (entry.take() == false)
                {
                    cursor += 1;
                    continue;
                }
                if (entry.getServerChannelInfo().isOpen() == false)
                {
                    logger.debug("通道{}已经关闭，不处理消息，直接退出", entry.getServerChannelInfo().getAddress());
                    return;
                }
                
                byte msgCommand = entry.getCommand();
                int handlerIndex = 0;
                for (; handlerIndex < limit; handlerIndex++)
                {
                    if (command[handlerIndex] == msgCommand)
                    {
                        break;
                    }
                }
                if (handlerIndex == limit)
                {
                    // 消息类型不识别，直接关闭
                    entry.getServerChannelInfo().close();
                    continue;
                }
                int interceptorIndex = 0;
                int result = -1;
                for (; interceptorIndex < interceptorNum;)
                {
                    result = inInterceptors[interceptorIndex].inInterceptor(entry);
                    interceptorIndex++;
                    if (result > 1)
                    {
                        break;
                    }
                }
                if (result == EntryInterceptor.closed)
                {
                    entry.getServerChannelInfo().close();
                    continue;
                }
                if (result != EntryInterceptor.interceptored)
                {
                    handlers[handlerIndex].handler(entry);
                }
                for (int i = interceptorNum - interceptorIndex; i < interceptorNum; i++)
                {
                    result = outInterceptors[i].outInterceptor(entry);
                    if (result > 1)
                    {
                        break;
                    }
                }
                if (result == EntryInterceptor.closed)
                {
                    entry.getServerChannelInfo().close();
                    continue;
                }
                ByteCache businessData = entry.getBusinessData();
                ByteBuffer writeBuffer = ByteBufferPool.getBuffer(businessData.getCount() + 8);
                headFactory.putHeadInBuffer(writeBuffer, entry.getCommand(), businessData.getCount());
                writeBuffer.put(entry.getActionType()).put(entry.getCommand()).put(entry.getRequestResult());
                writeBuffer.putInt(businessData.getCount());
                writeBuffer.put(businessData.getDirectArray(), 0, businessData.getCount());
                writeBuffer.flip();
                entry.getWritePacket().flowDone(writeBuffer);
                logger.debug("序号：{}已经处理完毕", cursor);
                if (entry.getServerChannelInfo().isOpen())
                {
                    entry.getWritePacket().tryWrite();
                }
                else
                {
                    logger.debug("通道{}已经关闭，不发送消息", entry.getServerChannelInfo().getAddress());
                }
            }
            catch (Exception e)
            {
                logger.error("消息处理过程发生异常，关闭通道", e);
                entry.getServerChannelInfo().close();
            }
        }
    }
    
    public long cursor()
    {
        return cursor;
    }
}
